package expertise.titan.datasource.impl;

import org.apache.commons.configuration.BaseConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;

import com.thinkaurelius.titan.core.TitanFactory;
import com.thinkaurelius.titan.core.TitanGraph;

import expertise.titan.datasource.TitanDataSource;

/**
 * This abstract class modularizes the configuration initialization for a
 * {@link TitanDataSource}. It sets cassandra specific options using spring
 * injected properties.
 * 
 * @author bdgould
 * 
 */
public abstract class ExpertiseClientConfigurer implements TitanDataSource, InitializingBean {

    private static final Logger log = Logger.getLogger(ExpertiseClientConfigurer.class);

    @Value("${expertise.titan.cassandra:false}")
    private boolean cassandra;

    @Value("${expertise.titan.cassandra.hosts:127.0.0.1}")
    private String[] hosts;

    @Value("${expertise.titan.cassandra.storage.port:9160}")
    private int storagePort;

    @Value("${expertise.titan.cassandra.storage.connection-timeout:5000}")
    private int connectionTimeout;

    @Value("${expertise.titan.cassandra.storage.connection-pool-size:32}")
    private int poolSize;

    @Value("${expertise.titan.cassandra.storage.replication-factor:1}")
    private int replicationFactor;

    @Value("${expertise.titan.local-dir:target/titan-data}")
    private String localDir;

    protected TitanGraph graph;

    @Override
    public void afterPropertiesSet() throws Exception {
        if (cassandra) {
            graph = TitanFactory.open(configureCassandraClient());
        } else {
            graph = TitanFactory.open(localDir);
        }
    }

    private Configuration configureCassandraClient() {
        log.info("Configuring Cassandra Titan client");
        Configuration conf = new BaseConfiguration();
        conf.addProperty("storage.backend", "cassandra");
        conf.addProperty("storage.keyspace", "expertise");
        conf.addProperty("storage.hostname", buildHostNameProperty());
        conf.addProperty("storage.port", storagePort);
        conf.addProperty("storage.connection-timeout", connectionTimeout);
        conf.addProperty("storage.connection-pool-size", poolSize);
        conf.addProperty("storage.replication-factor", replicationFactor);
        // conf.addProperty("storage.read-consistency-level", "");
        // conf.addProperty("storage.write-consistency-level", "");
        return conf;
    }

    private String buildHostNameProperty() {
        final StringBuffer buff = new StringBuffer();
        for (int i = 0; i < hosts.length; i++) {
            buff.append(i == 0 ? hosts[i] : "," + hosts[i]);
        }
        return buff.toString();
    }

}
